#include <pthread.h>
#include <thread>
#include <boost/bind.hpp>
#include <boost/asio.hpp>
#include <boost/timer/timer.hpp>
#include <boost/exception/all.hpp>
#include "error_message.hpp"

#ifndef _asio_ios_hpp
#define   _asio_ios_hpp

class Asio_ios
{
  public:
     //  程序内部定义时 io_service 必须在前面, 否则无法保证安全退出
     std::shared_ptr<boost::asio::io_service>    sp_ios;
     Asio_ios( unsigned int cpu_affinity_begin_number,  unsigned int int_thread_per_ios, bool bool_set_cpu_affinity );
     ~Asio_ios();
     void  start();
     void  set_display_level( unsigned char display_level_in );

  private:
    unsigned char d_display_level;
    unsigned int  d_int_concurrency;
    unsigned int  d_cpu_affinity_begin_number;
    unsigned int  d_int_thread_per_ios;
    volatile bool  d_bool_start;
    std::shared_ptr<boost::asio::io_service::work>  d_sp_work;
    std::vector<std::shared_ptr<std::thread> >  d_vt_sp_thread;
    void  run( std::shared_ptr<boost::asio::io_service>  sp_ios );
    void  set_cpu_affinity( std::shared_ptr<std::thread> sp_thread );
};


void  Asio_ios::set_display_level( unsigned char display_level_in )
{
  d_display_level = display_level_in;
  return;
}


void  Asio_ios::start()
{
  d_bool_start = true;
  return;
}


Asio_ios::Asio_ios( unsigned int cpu_affinity_begin_number=1, unsigned int int_thread_per_ios=1, bool bool_set_cpu_affinity=true  ):
  sp_ios( new boost::asio::io_service ),
  d_display_level( g_display_level_default ),
  d_cpu_affinity_begin_number( cpu_affinity_begin_number ),
  d_int_thread_per_ios( int_thread_per_ios ),
  d_bool_start( false ),
  d_sp_work( new boost::asio::io_service::work( *sp_ios )  )
{
  unsigned int  int_thread_count;

  d_int_concurrency = std::thread::hardware_concurrency();   // 硬件并发数
  if( 0==d_int_concurrency )   d_int_concurrency = 1;
  if( d_cpu_affinity_begin_number>=d_int_concurrency )  d_cpu_affinity_begin_number = 0;
  if( 0==int_thread_per_ios )   int_thread_per_ios = d_int_concurrency;
  int_thread_count =  int_thread_per_ios;

  //  创建多个线程, 做ASIO数据转发
  while( 1  )  {
    // ios绑定1个线程
    std::shared_ptr<std::thread>  sp_thread( new std::thread(std::bind( &Asio_ios::run, this,  sp_ios) ) );
    d_vt_sp_thread.push_back( sp_thread );
    if( bool_set_cpu_affinity )  set_cpu_affinity( sp_thread );
    --int_thread_per_ios;
    if( 0==int_thread_per_ios )  break;
    continue;
    }

  if( d_display_level>=1 )  log_message_int( "create thread count",  int_thread_count );
  return;
}


void Asio_ios::set_cpu_affinity( std::shared_ptr<std::thread> sp_thread )
{
  cpu_set_t   cpu_set1;
  CPU_ZERO( &cpu_set1 );
  CPU_SET( d_cpu_affinity_begin_number, &cpu_set1 );
  ++d_cpu_affinity_begin_number;
  if( d_cpu_affinity_begin_number>=d_int_concurrency )  d_cpu_affinity_begin_number = 0;
  if ( 0!=pthread_setaffinity_np( sp_thread->native_handle(), sizeof(cpu_set1), &cpu_set1 ) )    {
    if( d_display_level>=1 ) {
      std::string  str_msg   = "set cpu set affinity failed. cpu_bind_number=";
      str_msg += std::to_string( d_cpu_affinity_begin_number );
      error_message_str( str_msg );
      }
    return;
    }
  return;
}


Asio_ios::~Asio_ios()
{
  d_sp_work.reset();
  for( unsigned int i=0; i<d_vt_sp_thread.size(); ++i ) {
     d_vt_sp_thread[i]->join();
     continue;
     }
  if( d_display_level>=2 )   warn_message( "~Asio_ios." );
  return;
}


void Asio_ios::run( std::shared_ptr<boost::asio::io_service>  sp_ios )
{
  std::string str_msg;

  while(!d_bool_start)  {    // 休息等待主线程的初始化,
    usleep(1000);
    continue;
    }
  while(1)  {
    try {
      sp_ios->run();
      }
    catch(...)  {
      str_msg = boost::current_exception_diagnostic_information();
      error_message_str( str_msg );
      if( 1==d_int_thread_per_ios )   {  // 满足条件per_ios_thread_count =1,  才可以继续循环, 应该是多数情况都满足
        sp_ios->reset();
        continue;
        }
      break;
      }
    break;
    }
  return;
}

//#################################################################################
//  下面为开发过程中的测试程序,  已经验证了程序的正确性.
#ifdef DEBUG

void asio_ios_calc_sum(unsigned int id, unsigned long long int_max)
{
  std::string str_msg;
  unsigned long long sum;
  unsigned long long i;

  str_msg = "run";
  str_msg += std::to_string(id);
  str_msg += " begin. ";
  log_message_str(str_msg);
  sum = 0 ;
  for(i=0; i<=int_max; ++i )  sum += i;
  str_msg = "run";
  str_msg += std::to_string(id);
  str_msg += " end. sum=";
  str_msg += std::to_string(sum);
  log_message_str(str_msg);
  return;
}


void test_asio_ios(void)
{
  Asio_ios  ios1(1);
  Asio_ios  ios2(2);

  boost::asio::io_service &io_service1( *ios1.sp_ios );
  boost::asio::io_service &io_service2 ( *ios2.sp_ios );
  boost::timer::auto_cpu_timer   a2;

 io_service1.post(std::bind(asio_ios_calc_sum,1,100000));
 io_service1.post(std::bind(asio_ios_calc_sum,2,1000000000));
 io_service2.post(std::bind(asio_ios_calc_sum,3,100000000));
 io_service1.post(std::bind(asio_ios_calc_sum,4,10000000));
 io_service2.post(std::bind(asio_ios_calc_sum,5,100000000));
 io_service1.post(std::bind(asio_ios_calc_sum,6,100000));
 io_service1.post(std::bind(asio_ios_calc_sum,7,10000000));
 io_service1.post(std::bind(asio_ios_calc_sum,8,1000));
 io_service1.post(std::bind(asio_ios_calc_sum,9,1000000));
 io_service2.post(std::bind(asio_ios_calc_sum,10,10));

  log_message("test main thread start.");
  ios1.start();
  ios2.start();
  log_message("test main over.");
  return;
}
#endif // DEBUG

#endif  /* _asio_ios_hpp */